package main

import (
	"os"
	"github.com/scue/go-hello/pipeline"
	"bufio"
	"fmt"
	//"time"
)

func main() {
	src := "net_large.in"
	out := "net_large.out"
	size := 512000000
	//writeSourceFile(src, size)
	p := createNetworkPipeline(src, size, 8)

	writeToFile(p, out)
	printFile(out)

	//time.Sleep(time.Hour)
}

func printFile(fileName string) {
	file, err := os.Open(fileName)
	defer file.Close()
	if err != nil {
		panic(err)
	}

	p := pipeline.ReaderSource(file, pipeline.ChunkSizeUnlimited)

	n := 0
	for v := range p {
		if n > 100 {
			break
		}
		fmt.Println(v)
		n++
	}
}

func writeToFile(source <-chan int, fileName string) {
	file, err := os.Create(fileName)
	defer file.Close()

	if err != nil {
		panic(err)
	}

	writer := bufio.NewWriter(file)
	defer writer.Flush()
	pipeline.WriterSink(writer, source)
}

func createNetworkPipeline(filename string, fileSize, chunkCount int) <-chan int {
	// FIXME fileSize可以自行获取
	// FIXME 处理非整除的情况
	chunkSize := fileSize / chunkCount
	results := make([]string, 0)

	for i := 0; i < chunkCount; i++ {
		file, err := os.Open(filename)
		if err != nil {
			panic(err)
		}
		file.Seek(int64(i*chunkSize), 0)
		source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
		addr := fmt.Sprintf(":%d", 7000+i)
		pipeline.NetworkSink(addr, source)
		results = append(results, addr)
		// FIXME 关闭文件，或是返回给其他函数
		// file.Close()
	}

	sortResults := make([]<-chan int, 0)
	for _, addr := range results {
		sortResults = append(sortResults, pipeline.InMemSort(pipeline.NetworkSource(addr)))
	}

	return pipeline.MergeN(sortResults...)
}

func createPipeline(filename string, fileSize, chunkCount int) <-chan int {
	// FIXME fileSize可以自行获取
	// FIXME 处理非整除的情况
	chunkSize := fileSize / chunkCount
	results := make([]<-chan int, 0)
	for i := 0; i < chunkCount; i++ {
		file, err := os.Open(filename)
		if err != nil {
			panic(err)
		}
		file.Seek(int64(i*chunkSize), 0)
		source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
		results = append(results, pipeline.InMemSort(source))
		// FIXME 关闭文件，或是返回给其他函数
		// file.Close()
	}

	return pipeline.MergeN(results...)
}

func writeSourceFile(fileName string, size int) {
	file, err := os.Create(fileName)
	defer file.Close()
	if err != nil {
		panic(err)
	}
	p := pipeline.RandomSource(size / 8)

	// 使用bufio自带的缓冲区，减少数据频繁的读写
	writer := bufio.NewWriter(file)
	pipeline.WriterSink(writer, p)
	writer.Flush()
}
